package com.atguigu.cdc;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Properties;

/**
 * Flink CDC DataStream 方式的应用
 *
 * @author lvbingbing
 * @date 2022-03-18 13:08
 */
public class FlinkDataStreamCDC {
    public static void main(String[] args) throws Exception {
        // 1. 准备流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 开启检查点
        // flinkCDC 将读取 binlog 的位置信息以状态的方式保存在 CK ，如果想要做到断点续传，需要从 checkpoint 或者 savepoint 启动程序
        // 2.1 开启Checkpoint，每隔 5s 做一次 CK，并指定 CK 的一致性语义
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 2.2 设置超时时间为 1分钟
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 2.3 指定从 CK 自动重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, 2000L));
        // 2.4 设置任务关闭的时候保留最后一次 CK 数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));
        // 2.6 设置访问 hdfs 的用户名
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        // 3. 创建 Flink-MySQL-CDC 的 Source
        Properties props = new Properties();
        props.setProperty("scan.startup.mode", "initial");
        SourceFunction<String> sourceFunction = MySQLSource.<String>builder()
                .hostname("hadoop102")
                .port(3306)
                .username("root")
                .password("139559")
                .databaseList("gmall2021_realtime")
                // 可选配置项，如果不指定该参数，则会读取上一个配置中指定的数据库下的所有表的数据
                // 注意：指定的时候需要使用 "db.table" 的方式
                .tableList("gmall2021_realtime.t_user")
                .debeziumProperties(props)
                .deserializer(new StringDebeziumDeserializationSchema())
                .build();
        // 4. 使用 CDC Source 从 MySQL 读取数据
        DataStreamSource<String> mysqlDS = env.addSource(sourceFunction);
        // 5. 打印输出
        mysqlDS.print();
        // 6. 执行任务
        env.execute();
    }
}
